It seems like this only receives txs upon mempool reorgs. I would have thought that the bloom filter should update as soon as we add the tx to the pool to avoid unnecessary gossip

I've demonstrated in TestTxPoolPopulatesBloomFilter() that it will send new txs as well and that the filter receives them.

Sorry, my original comment wasn't questioning whether or not the txs would get added to the bloom filter eventually. My comment was questioning whether or not it is smart to wait until a mempool reorg happens to populate the bloom filter.

In all of our current mempool implementations, we include the transactions into the bloom filter prior to returning from Add.

This only seems to add the newly issued transactions into the bloom filter, don't we need to reintroduce all the transactions into the bloom filter?

Yup; this is now done in NewSet() and tested in TestTxPoolPopulatesBloomFilter().

While I think it is good that we initialize the bloom filter in NewSet, I was referring to populating the bloom filter with existing transactions in the mempool after we reset the bloom filter.

ev.Txs doesn't include transactions that were previously in the mempool. So whenever we reset the bloom filter, we are going to start receiving (permanently, until those txs are removed from the mempool) duplicate tx gossip.

It is probably more robust to use a multiple of the current mempool size as a lower bound here. Otherwise we could get into a situation where we end up resetting every time we see a new tx

Ahh, nice idea. I had no idea how to set this (hence making it an argument). I've arbitrarily gone with 2x pool size but could always reintroduce an argument to configure that.

I feel like this makes the "receive" followed by "send" logic a bit more clear

I have a strong dislike of the "variable look-back" pattern as it causes a non-linearity in reading the code.

I'm being hyperbolic here, but this is the cognitive pattern I'm trying to avoid:

1. Push b to my brain stack without knowing how it will be used (context aids memory).
2. Set some b, making sure to notice that it's = and not :=, recalling my earlier stack push.
3. Parse some unrelated "distraction" code, keeping b at the back of my mind.
4. Use b.

vs

1. Declare and set b, pushing it to my brain stack but with context as to where it came from.
2. Use b.

Granted in this particular solution it's not all that significant because the "distraction" code is minimal, but I still feel like it's an anti-pattern. All that said, I'm open to having my mind changed if you think there are sufficient reasons to change.

fwiw, using := rather than = won't compile.

Would you be okay with some helper functions instead? Something like:

func (p *pipe) loop() {
	defer close(p.done)
	for {
		b, ok := p.read()
		if !ok {
			return
		}

		if !p.write(b) {
			return
		}
	}
}

func (p *pipe) read() (*types.Block, bool) {
	select {
	case b := <-p.in:
		return b, true
	case <-p.quit:
		return nil, false
	}
}

func (p *pipe) write(b *types.Block) bool {
	select {
	case p.out <- core.ChainHeadEvent{Block: b}:
		return true
	case <-p.quit:
		return false
	}
}

It's much more verbose... But it doesn't have your (perhaps limited to you :wink) meat stack issue

So, with this PR the mempool's validity checking is based on the settled state right?

Intent is currently undefined. Sorry, I should have been clearer in what I meant in the PR description (i.e. I'm kicking that can down the road entirely):

Any specific rules for when to reject a tx from the mempool will likely require some libevm mods so can follow later.

The p2p architecture and gossip was all so new to me that I found this PR relatively difficult, so wanted to focus on learning it and getting the wiring correct. I'll concentrate on things like validity when we have the VM to orchestrate all the moving parts.